【Redis】Redis Cluster初始化及PING消息的发送

Posted on

Cluster消息类型定义

1
2
3
4
#define CLUSTERMSG_TYPE_PING 0          /* Ping消息类型，节点间进行通信交换信息的消息 */
#define CLUSTERMSG_TYPE_PONG 1          /* Pong消息类型 (Ping命令的回复) */
#define CLUSTERMSG_TYPE_MEET 2          /* Meet消息类型，表示节点加入集群 */
#define CLUSTERMSG_TYPE_FAIL 3          /* FAIL消息类型，表示节点下线*/

在Redis初始化服务initServer函数中，调用aeCreateTimeEvent注册了时间事件，周期性的执行serverCron函数，在serverCron中可以看到每隔100ms调用一次clusterCron函数，执行Redis Cluster定时任务：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
void initServer(void) {
    
    // 省略...
    if (aeCreateTimeEvent(server.el, 1, serverCron, NULL, NULL) == AE_ERR) {
        serverPanic("Can't create event loop timers.");
        exit(1);
    }
    // 省略...
}

int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
    // 省略...
    
    /* Redis Cluster 定时任务，每隔100ms调用一次 */
    run_with_period(100) {
        if (server.cluster_enabled) clusterCron();
    }
    
    // 省略...
}

clusterCron

clusterCron是集群相关的定时执行函数，每100ms执行一次：

  1. 遍历集群中的所有节点，校验是否有连接中断的节点并进行重新连接
    • 如果节点是自身或者是没有地址的节点，跳过
    • 如果节点处于握手状态并且已经超时，跳过
    • 如果连接为空，调用connConnect进行连接，回调函数为clusterLinkConnectHandler
  2. 每执行10次clusterCron函数时随机选取五个节点，然后从这五个节点选出最早收到PONG回复的那个节点，也就是找出最久没有进行通信的那个节点，向其发送PING消息，clusterCron每100ms执行一次，执行10次是1000ms，也就是说每1秒选取一个节点调用clusterSendPing函数发送一次PING消息
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
void clusterCron(void) {
    
    // ...
    
    /* 校验是否有连接中断的节点并进行重新连接 */
    di = dictGetSafeIterator(server.cluster->nodes);
    server.cluster->stats_pfail_nodes = 0;
    // 遍历集群中的所有节点
    while((de = dictNext(di)) != NULL) {
        // 获取集群节点
        clusterNode *node = dictGetVal(de);

        /* 如果节点是自身或者是没有地址的节点，跳过 */
        if (node->flags & (CLUSTER_NODE_MYSELF|CLUSTER_NODE_NOADDR)) continue;

        if (node->flags & CLUSTER_NODE_PFAIL)
            server.cluster->stats_pfail_nodes++;

        /* 如果节点处于握手状态并且已经超时 */
        if (nodeInHandshake(node) && now - node->ctime > handshake_timeout) {
            clusterDelNode(node);
            continue;
        }

        if (node->link == NULL) {
            // 创建clusterLink
            clusterLink *link = createClusterLink(node);
            link->conn = server.tls_cluster ? connCreateTLS() : connCreateSocket();
            connSetPrivateData(link->conn, link);
            // 建立连接，监听函数为clusterLinkConnectHandler
            if (connConnect(link->conn, node->ip, node->cport, NET_FIRST_BIND_ADDR,
                        clusterLinkConnectHandler) == -1) {
                if (node->ping_sent == 0) node->ping_sent = mstime();
                serverLog(LL_DEBUG, "Unable to connect to "
                    "Cluster Node [%s]:%d -> %s", node->ip,
                    node->cport, server.neterr);

                freeClusterLink(link);
                continue;
            }
            // 设置link
            node->link = link;
        }
    }
    dictReleaseIterator(di);
    
    /* 每执行10次clusterCron函数，发送一次PING消息 */
    if (!(iteration % 10)) {
        int j;

        /* 随机选取节点并找到最早收到pong消息的节点 */
        for (j = 0; j < 5; j++) {
            // 随机选取节点
            de = dictGetRandomKey(server.cluster->nodes);
            // 获取节点
            clusterNode *this = dictGetVal(de);

            /* 如果节点的连接已中断或者本次PING命令处于活跃状态 */
            if (this->link == NULL || this->ping_sent != 0) continue;
            if (this->flags & (CLUSTER_NODE_MYSELF|CLUSTER_NODE_HANDSHAKE))
                continue;
            // 查找最早收到PONG消息的那个节点
            if (min_pong_node == NULL || min_pong > this->pong_received) {
                min_pong_node = this;
                min_pong = this->pong_received;
            }
        }
        // 如果是最早收到PONG消息的节点
        if (min_pong_node) {
            serverLog(LL_DEBUG,"Pinging node %.40s", min_pong_node->name);
            // 发送PING消息
            clusterSendPing(min_pong_node->link, CLUSTERMSG_TYPE_PING);
        }
    }
    
    // ...
}

clusterNode

clusterNode是集群中节点对应的结构体，包含了以下内容：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
typedef struct clusterNode {
    mstime_t ctime; /* 节点对象创建时间 */
    char name[CLUSTER_NAMELEN]; /* 节点名称 */
    int flags;      /* 节点标识 */
    uint64_t configEpoch; /* configEpoch */
    unsigned char slots[CLUSTER_SLOTS/8]; /* 节点负责的slots */
    sds slots_info; /* Slots信息 */
    int numslots;   /* 节点负责的slots数量 */
    int numslaves;  /* 从节点的数量 */
    struct clusterNode **slaves; /* 指向从节点的指针 */
    struct clusterNode *slaveof; /* 指向主节点的指针 */
    mstime_t ping_sent;      /* 最近一次发送PING消息的时间 */
    mstime_t pong_received;  /* 收到pong消息的时间 */
    mstime_t data_received;  /* Unix time we received any data */
    mstime_t fail_time;      /* 标记FAIL状态的时间 */
    mstime_t voted_time;     /* 最近一次投票的时间 */
    mstime_t repl_offset_time;  /* 收到主从复制offset的时间*/
    mstime_t orphaned_time;     /* Starting time of orphaned master condition */
    long long repl_offset;      /* 节点主从复制offset */
    char ip[NET_IP_STR_LEN];  /* IP */
    int port;                   /* 客户端通信端口 */
    int pport;                  /* 使用TLS协议的端口 */
    int cport;                  /* 集群通信端口 */
    clusterLink *link;          /* TCP/IP连接相关信息 */
    list *fail_reports;         /* List of nodes signaling this as failing */
} clusterNode;

clusterLinkConnectHandler

clusterLinkConnectHandler是建立连接的监听函数，当连接建立时会调用clusterLinkConnectHandler进行处理，在clusterLinkConnectHandler函数中可以看到，又调用了connSetReadHandler注册了可读事件的监听，对应的回调函数为clusterReadHandler，当收到其他节点发送的通信消息时会调用clusterReadHandler函数处理：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
void clusterLinkConnectHandler(connection *conn) {
    clusterLink *link = connGetPrivateData(conn);
    clusterNode *node = link->node;

    /* 校验连接是否成功 */
    if (connGetState(conn) != CONN_STATE_CONNECTED) {
        serverLog(LL_VERBOSE, "Connection with Node %.40s at %s:%d failed: %s",
                node->name, node->ip, node->cport,
                connGetLastError(conn));
        freeClusterLink(link);
        return;
    }

    /* 注册readHandler，监听函数为clusterReadHandler */
    connSetReadHandler(conn, clusterReadHandler);

    // 省略..
}

集群间通信

通信消息结构体定义

clusterMsg

集群间通信的消息对应的结构体为clusterMsg，里面包含了消息类型、发送消息节点的slots信息以及节点间通信的消息体clusterMsgData等信息：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
typedef struct {
    char sig[4];        /*  "RCmb"签名 */
    uint32_t totlen;    /* 消息总长度 */
    uint16_t ver;       /* 协议版本, 当前设置为1 */
    uint16_t port;      /* 端口 */
    uint16_t type;      /* 消息类型 */
    
    // 省略...
  
    char sender[CLUSTER_NAMELEN]; /* 发送消息节点的名称 */
    unsigned char myslots[CLUSTER_SLOTS/8]; /* 发送消息节点的slots信息 */
    char slaveof[CLUSTER_NAMELEN];
    char myip[NET_IP_STR_LEN];    /* 发送消息节点的ip */
    char notused1[32];  /* 32字节的保留数据 */
    uint16_t pport;      /* 使用TLS协议时的端口 */
    uint16_t cport;      /* 发送消息节点的集群总线端口，也就是用于集群间通信的端口 */
    uint16_t flags;      /* 发送消息节点的flags标识 */
    unsigned char state; /* 发送消息节点的集群状态 */
    unsigned char mflags[3]; /* Message flags: CLUSTERMSG_FLAG[012]_... */
    union clusterMsgData data; // 集群通信的实际消息
} clusterMsg;

clusterMsgData

clusterMsgData里面存储了节点间进行通信的实际消息，不同消息类型对应不同的数据结构：

  1. clusterMsgDataGossip：PING, MEET 和 PONG消息对应的数据结构
  2. clusterMsgDataFail：FAIL消息对应的数据结构
  3. clusterMsgDataPublish：PUBLISH消息对应的数据结构
  4. clusterMsgDataUpdate：UPDATE消息对应的数据结构
  5. clusterMsgModule：MODULE消息对应的数据结构
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
union clusterMsgData {
    /* PING, MEET and PONG消息对应的数据结构 */
    struct {
        clusterMsgDataGossip gossip[1];
    } ping;

    /* FAIL消息对应的数据结构 */
    struct {
        clusterMsgDataFail about;
    } fail;

    /* PUBLISH消息对应的数据结构 */
    struct {
        clusterMsgDataPublish msg;
    } publish;

    /* UPDATE消息对应的数据结构 */
    struct {
        clusterMsgDataUpdate nodecfg;
    } update;

    /* MODULE消息对应的数据结构 */
    struct {
        clusterMsgModule msg;
    } module;
};

clusterMsgDataGossip

clusterMsgDataGossip是集群间发送PING、MEET 和 PONG消息对应的数据结构，里面包含以下信息：

1
2
3
4
5
6
7
8
9
10
11
typedef struct {
    char nodename[CLUSTER_NAMELEN]; /* 节点名称 */
    uint32_t ping_sent; /* 发送PING命令的时间 */
    uint32_t pong_received; /* 收到PONG命令的时间 */
    char ip[NET_IP_STR_LEN];  /* 节点的IP */
    uint16_t port;              /* 用于客户端通信的端口 */
    uint16_t cport;             /* 集群间通信的端口 */
    uint16_t flags;             /* 节点的flags标识 */
    uint16_t pport;             /* 使用TLS协议时的端口 */
    uint16_t notused1;
} clusterMsgDataGossip;

PING消息的发送

clusterSendPing

clusterSendPing函数用于向指定节点发送PING消息，Ping消息中不仅包含当前节点的信息，也会随机选取一些其他的节点，将其他节点的信息封装在消息体中进行发送，随机选取节点的个数计算规则如下：

  • wanted：随机选取的节点个数，默认是集群中节点的数量除以10
  • freshnodes：随机选取的节点个数的最大值，默认集群中节点的数量减2

如果wanted小于3，那么将wanted置为3，也就是最少选取3个节点；

如果wanted大于freshnodes，将wanted置为freshnodes的值，也就是最大可以选取freshnodes个节点；

选取的节点个数wanted确定之后，处理逻辑如下：

  1. 调用clusterBuildMessageHdr函数构建消息头

  2. 根据wanted的数量随机选取节点，处于以下几种情况的节点将被跳过

    • FAIL下线状态的节点

    • 处于握手状态的节点

    • 没有地址信息的节点

    • 失去连接的节点并且没有配置slots.

  3. 调用clusterSetGossipEntry函数将选取的节点信息加入到消息体中

  4. 调用clusterSendMessage函数发送消息

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
void clusterSendPing(clusterLink *link, int type) {
    unsigned char *buf; /* 发送的消息数据*/
    clusterMsg *hdr; /* 节点间通信消息 */
    int gossipcount = 0; 
    int wanted; /* 选取的节点个数 */
    int totlen; /* 总长度 */

    // 集群中节点的数量 - 2
    int freshnodes = dictSize(server.cluster->nodes)-2;
    // 集群中节点的数量除以10
    wanted = floor(dictSize(server.cluster->nodes)/10);
    // 如果wanted小于3，则设置为3
    if (wanted < 3) wanted = 3;
    // 如果大于最大节点数，设置为freshnodes
    if (wanted > freshnodes) wanted = freshnodes;
  
    //...
  
    if (totlen < (int)sizeof(clusterMsg)) totlen = sizeof(clusterMsg);
    buf = zcalloc(totlen); // 分配空间
    hdr = (clusterMsg*) buf;
    if (link->node && type == CLUSTERMSG_TYPE_PING)
        link->node->ping_sent = mstime();// 更新发送PING消息时间
    // 构建消息头
    clusterBuildMessageHdr(hdr,type);
    
    /* 计算 gossip  */
    int maxiterations = wanted*3;
    while(freshnodes > 0 && gossipcount < wanted && maxiterations--) {
        // 随机选取节点
        dictEntry *de = dictGetRandomKey(server.cluster->nodes);
        clusterNode *this = dictGetVal(de);

        /* 如果是自身 */
        if (this == myself) continue;

        /* 如果是FAIL状态，跳过 */
        if (this->flags & CLUSTER_NODE_PFAIL) continue;

        /* 以下节点跳过:
         * 1) 处于握手状态的节点.
         * 3) 没有地址信息的节点.
         * 4) 失去连接的节点并且没有配置slots.
         */
        if (this->flags & (CLUSTER_NODE_HANDSHAKE|CLUSTER_NODE_NOADDR) ||
            (this->link == NULL && this->numslots == 0))
        {
            freshnodes--; /* Technically not correct, but saves CPU. */
            continue;
        }

        /* 如果节点已经添加 */
        if (clusterNodeIsInGossipSection(hdr,gossipcount,this)) continue;

        /* 添加到消息体中 */
        clusterSetGossipEntry(hdr,gossipcount,this);
        freshnodes--;
        gossipcount++;
    }
    // ...
    
    totlen = sizeof(clusterMsg)-sizeof(union clusterMsgData);
    totlen += (sizeof(clusterMsgDataGossip)*gossipcount);
    hdr->count = htons(gossipcount);
    hdr->totlen = htonl(totlen);
    // 发送消息
    clusterSendMessage(link,buf,totlen);
    zfree(buf);

}

构建消息头

clusterBuildMessageHdr

clusterBuildMessageHdr函数用于构建消息头，设置了消息发送者的节点相关信息：

  1. 设置了签名、消息类型、节点IP、端口等信息
  2. 设置发送消息节点的slots信息，如果发送消息的节点是从节点，需要使用它对应的主节点的slots信息
  3. 计算集群消息的总长度totlen，并设置到消息头中
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
void clusterBuildMessageHdr(clusterMsg *hdr, int type) {
    int totlen = 0;
    uint64_t offset;
    clusterNode *master;

    /* 如果是从节点, 使用它对应的主节点的信息 */
    master = (nodeIsSlave(myself) && myself->slaveof) ?
              myself->slaveof : myself;

    memset(hdr,0,sizeof(*hdr));
    hdr->ver = htons(CLUSTER_PROTO_VER);
    // 设置签名
    hdr->sig[0] = 'R';
    hdr->sig[1] = 'C';
    hdr->sig[2] = 'm';
    hdr->sig[3] = 'b';
    // 设置消息类型
    hdr->type = htons(type);
    memcpy(hdr->sender,myself->name,CLUSTER_NAMELEN);
    memset(hdr->myip,0,NET_IP_STR_LEN);
    if (server.cluster_announce_ip) {
        // 设置ip
        strncpy(hdr->myip,server.cluster_announce_ip,NET_IP_STR_LEN);
        hdr->myip[NET_IP_STR_LEN-1] = '\0';
    }

    /* 处理端口 */
    int announced_port, announced_pport, announced_cport;
    deriveAnnouncedPorts(&announced_port, &announced_pport, &announced_cport);
    // 设置当前节点的slots信息
    memcpy(hdr->myslots,master->slots,sizeof(hdr->myslots));
    memset(hdr->slaveof,0,CLUSTER_NAMELEN);
    if (myself->slaveof != NULL)
        memcpy(hdr->slaveof,myself->slaveof->name, CLUSTER_NAMELEN);
    // 设置端口
    hdr->port = htons(announced_port);
    hdr->pport = htons(announced_pport);
    hdr->cport = htons(announced_cport);
    // 设置标识
    hdr->flags = htons(myself->flags);
    // 设置集群状态
    hdr->state = server.cluster->state;

    /* 设置currentEpoch和configEpoch */
    hdr->currentEpoch = htonu64(server.cluster->currentEpoch);
    hdr->configEpoch = htonu64(master->configEpoch);

    /* 设置主从复制的offset. */
    if (nodeIsSlave(myself))
        offset = replicationGetSlaveOffset();
    else
        offset = server.master_repl_offset;
    hdr->offset = htonu64(offset);

    if (nodeIsMaster(myself) && server.cluster->mf_end)
        hdr->mflags[0] |= CLUSTERMSG_FLAG0_PAUSED;

    /* 计算消息总长度 */
    if (type == CLUSTERMSG_TYPE_FAIL) {
        totlen = sizeof(clusterMsg)-sizeof(union clusterMsgData);
        totlen += sizeof(clusterMsgDataFail);
    } else if (type == CLUSTERMSG_TYPE_UPDATE) {
        totlen = sizeof(clusterMsg)-sizeof(union clusterMsgData);
        totlen += sizeof(clusterMsgDataUpdate);
    }
    // 设置消息总长度
    hdr->totlen = htonl(totlen);
}

构建消息体

clusterSetGossipEntry

clusterSetGossipEntry函数用于构建消息体，将随机选取的其他节点信息加入到ping消息对应的数组hdr->data.ping.gossip[i]中，并设置节点的相关信息：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
void clusterSetGossipEntry(clusterMsg *hdr, int i, clusterNode *n) {
    clusterMsgDataGossip *gossip;
    gossip = &(hdr->data.ping.gossip[i]);
    memcpy(gossip->nodename,n->name,CLUSTER_NAMELEN);
    // 设置PING消息发送时间
    gossip->ping_sent = htonl(n->ping_sent/1000);
    // 设置收到PONG消息时间
    gossip->pong_received = htonl(n->pong_received/1000);
    // 设置IP
    memcpy(gossip->ip,n->ip,sizeof(n->ip));
    // 设置端口
    gossip->port = htons(n->port);
    // 设置集群端口
    gossip->cport = htons(n->cport);
    // 设置标识
    gossip->flags = htons(n->flags);
    gossip->pport = htons(n->pport);
    gossip->notused1 = 0;
}

PING消息的处理

clusterReadHandler

由上面的clusterLinkConnectHandler函数可知，收到其他节点发送的通信消息时会调用clusterReadHandler函数处理，在clusterReadHandler函数中会开启while循环，不断读取数据，直到获取完整的数据（收到的数据长度rcvbuflen等于消息中设置数据总长度时），调用clusterProcessPacket函数处理收到的消息：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
void clusterReadHandler(connection *conn) {
    clusterMsg buf[1];
    ssize_t nread;
    clusterMsg *hdr;
    clusterLink *link = connGetPrivateData(conn);
    unsigned int readlen, rcvbuflen;

    while(1) { 
        rcvbuflen = link->rcvbuf_len;
        
        // 省略...
      
        // 读取数据
        nread = connRead(conn,buf,readlen);
        // 如果数据读取完毕
        if (nread == -1 && (connGetState(conn) == CONN_STATE_CONNECTED)) return; 
        
        // 省略...
        
        /* 如果已经获取完整数据（rcvbuflen等于消息中设置数据总长度），处理数据包 */
        if (rcvbuflen >= 8 && rcvbuflen == ntohl(hdr->totlen)) {
            // 处理消息
            if (clusterProcessPacket(link)) { 
                if (link->rcvbuf_alloc > RCVBUF_INIT_LEN) {
                    zfree(link->rcvbuf);
                    link->rcvbuf = zmalloc(link->rcvbuf_alloc = RCVBUF_INIT_LEN);
                }
                link->rcvbuf_len = 0;
            } else {
                return; 
            }
        }
    }
}

clusterProcessPacket

clusterProcessPacket函数用于处理收到的通信消息，可以看到有许多if else分支，根据消息类型的不同，进行了不同的处理，这里先只关注PING消息的处理：

  1. 如果消息类型是PING或者MEET，调用clusterSendPing函数发送PONG消息，传入的消息类型为CLUSTERMSG_TYPE_PONG，说明PING和PONG消息都是通过clusterSendPing函数实现的，PING和PONG消息的数据结构一致，那么回复的PONG消息中也会带上回复者的节点信息以及回复者随机选取的其他节点信息，以此达到节点间交换信息的目的
  2. 如果是PING, PONG或者MEET消息，并且sender不为空，不为空表示发送消息的节点是当前节点已知的，调用clusterProcessGossipSection函数处理消息体中的Gossip数据
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
int clusterProcessPacket(clusterLink *link) {
    // 获取发送的消息
    clusterMsg *hdr = (clusterMsg*) link->rcvbuf;
    // 消息长度
    uint32_t totlen = ntohl(hdr->totlen);
    // 消息类型
    uint16_t type = ntohs(hdr->type);
    mstime_t now = mstime();
    uint16_t flags = ntohs(hdr->flags);
    uint64_t senderCurrentEpoch = 0, senderConfigEpoch = 0;
    clusterNode *sender;

    // 省略...
  
    /* 校验发送者是否是已知的节点 */
    sender = clusterLookupNode(hdr->sender);

    /* 更是发送者收到数据的时间*/
    if (sender) sender->data_received = now;
   
    // 省略...
   
    /* 如果是PING消息或者MEET消息 */
    if (type == CLUSTERMSG_TYPE_PING || type == CLUSTERMSG_TYPE_MEET) {
        serverLog(LL_DEBUG,"Ping packet received: %p", (void*)link->node);
      
        if (!sender && type == CLUSTERMSG_TYPE_MEET)
            clusterProcessGossipSection(hdr,link);

        /* 发送PONG消息，这里传入的类型是CLUSTERMSG_TYPE_PONG */
        clusterSendPing(link,CLUSTERMSG_TYPE_PONG);
    }

    /* PING, PONG, MEET 消息 */
    if (type == CLUSTERMSG_TYPE_PING || type == CLUSTERMSG_TYPE_PONG ||
        type == CLUSTERMSG_TYPE_MEET)
    {
      
        // 省略...

        /* 处理消息体中的Gossip节点数据 */
        if (sender) clusterProcessGossipSection(hdr,link);
    } else {
        serverLog(LL_WARNING,"Received unknown packet type: %d", type);
    }
    return 1;
}

clusterProcessGossipSection

clusterProcessGossipSection函数用于处理clusterMsg中的Gossip节点信息g，它从集群消息中获取Gossip节点数据，根据节点数量进行遍历：

  1. 调用clusterLookupNode函数根据nodename从当前收到消息的节点的集群中查找Gossip节点，查找结果记为node

    • 如果node如果不为空，说明可以从当前节点的集群中找到，Gossip节点针对当前节点是已知的，需要注意node指向的是当前收到消息节点中维护的相同nodename的节点，g指向当前正在遍历的gossip节点（sender发送的消息中携带gossip数组），注意两者的区别
    • 如果node如果为空，说明Gossip节点针对当前节点是未知的，之前不在当前节点维护的集群节点中

  2. 如果node不为空，也就是当前收到消息这个节点的集群中已经存在node节点，进行如下处理：

    （1）发送消息的节点sender是主节点时有以下两种情况：

    • 如果node是FAIL或者PFAIL状态，需要将sender加入到node节点的下线链表fail_reports中，表示sender认为node节点下线（clusterNodeAddFailureReport函数）
    • 断是否有必要将node标记为下线状态（markNodeAsFailingIfNeeded函数）
    • 如果node不是FAIL或者PFAIL状态，需要校验node是否已经在sender的下线节点链表fail_reports中，如果在需要从中移除

    （2）如果node节点不是FAIL、PFAIL、NOADDR状态，并且node的ip或者端口与g指向的gossip节点中的ip或者端口不一致，需要更新node中的ip和端口

  3. 如果node为空，说明之前不在当前节点维护的集群节点中，如果gossip节点不处于NOADDR状态并且不在nodes_black_list中，新建节点，加入到当前收到消息的节点维护的集群数据server.cluster中

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
void clusterProcessGossipSection(clusterMsg *hdr, clusterLink *link) {
    uint16_t count = ntohs(hdr->count);
    // 获取clusterMsgDataGossip数据
    clusterMsgDataGossip *g = (clusterMsgDataGossip*) hdr->data.ping.gossip;
    // 发送消息的节点
    clusterNode *sender = link->node ? link->node : clusterLookupNode(hdr->sender);

    while(count--) {
        // 获取节点标识
        uint16_t flags = ntohs(g->flags);
        clusterNode *node;
        sds ci;

        if (server.verbosity == LL_DEBUG) {
            ci = representClusterNodeFlags(sdsempty(), flags);
            serverLog(LL_DEBUG,"GOSSIP %.40s %s:%d@%d %s",
                g->nodename,
                g->ip,
                ntohs(g->port),
                ntohs(g->cport),
                ci);
            sdsfree(ci);
        }

        /* 根据nodename查找节点，node指向当前收到消息节点中维护的节点*/
        node = clusterLookupNode(g->nodename);
        // 如果节点已知
        if (node) {
            /* 如果发送者是主节点 */
            if (sender && nodeIsMaster(sender) && node != myself) {
                // 如果gossip节点是FAIL或者PFAIL状态
                if (flags & (CLUSTER_NODE_FAIL|CLUSTER_NODE_PFAIL)) {
                    // 将sender加入到node节点的下线链表fail_reports中，表示sender认为node节点下线
                    if (clusterNodeAddFailureReport(node,sender)) {
                        serverLog(LL_VERBOSE,
                            "Node %.40s reported node %.40s as not reachable.",
                            sender->name, node->name);
                    }
                    // 判断是否有必要将节点置为客观下线
                    markNodeAsFailingIfNeeded(node);
                } else {
                    // 校验节点是否在下线节点链表fail_reports中，如果在需要移除恢复在线状态
                    if (clusterNodeDelFailureReport(node,sender)) {
                        serverLog(LL_VERBOSE,
                            "Node %.40s reported node %.40s is back online.",
                            sender->name, node->name);
                    }
                }
            }

            /* 如果节点不是FAIL或者PFAIL状态，并且node中记录的ping发送时间为0，并且node不在fail_reports中*/
            if (!(flags & (CLUSTER_NODE_FAIL|CLUSTER_NODE_PFAIL)) &&
                node->ping_sent == 0 &&
                clusterNodeFailureReportsCount(node) == 0)
            {
                mstime_t pongtime = ntohl(g->pong_received);
                pongtime *= 1000; /* 转为毫秒 */
                if (pongtime <= (server.mstime+500) &&
                    pongtime > node->pong_received)
                {
                    node->pong_received = pongtime; // 更新收到pong消息时间
                }
            }

            /* 如果node节点不是FAIL、PFAIL、NOADDR状态，并且node的ip或者端口与g节点中的ip或者端口不一致，需要更新node中的ip和端口 */
            /* 需要注意node节点和g节点的区别，node节点是从当前收到消息节点中根据节点id查找到的节点，也就是接收者自己记录的节点信息 */
            /* g指向当前在遍历的那个gossip节点，也就是发送者带过来的节点信息 */
            if (node->flags & (CLUSTER_NODE_FAIL|CLUSTER_NODE_PFAIL) &&
                !(flags & CLUSTER_NODE_NOADDR) &&
                !(flags & (CLUSTER_NODE_FAIL|CLUSTER_NODE_PFAIL)) &&
                (strcasecmp(node->ip,g->ip) ||
                 node->port != ntohs(g->port) ||
                 node->cport != ntohs(g->cport)))
            {
                if (node->link) freeClusterLink(node->link);
                // 更新node节点中的端口、ip信息
                memcpy(node->ip,g->ip,NET_IP_STR_LEN);
                node->port = ntohs(g->port);
                node->pport = ntohs(g->pport);
                node->cport = ntohs(g->cport);
                node->flags &= ~CLUSTER_NODE_NOADDR;
            }
        } else { // 如果节点未知
            /* 如果节点不处于NOADDR状态并且不在nodes_black_list中 */
            if (sender &&
                !(flags & CLUSTER_NODE_NOADDR) &&
                !clusterBlacklistExists(g->nodename))
            {
                clusterNode *node;
                // 创建节点
                node = createClusterNode(g->nodename, flags);
                memcpy(node->ip,g->ip,NET_IP_STR_LEN);
                node->port = ntohs(g->port);
                node->pport = ntohs(g->pport);
                node->cport = ntohs(g->cport);
                // 加入到当前节点维护的集群server.cluster中
                clusterAddNode(node);
            }
        }

        /* 遍历下一个节点 */
        g++;
    }
}

总结

参考

极客时间 - Redis源码剖析与实战(蒋德钧)
zhaiguanjie-Redis源码剖析

Redis版本：redis-6.2.5